package trouger.dmt;

import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.TaggedInputSplitWrapper;
import org.apache.hadoop.util.ReflectionUtils;

class BaseMapper<T extends MapperEntity<K, V>, K, V>
extends MapReduceBase
implements Mapper<LongWritable, Text, K, V>{
	
	
	protected int sourceCount;
	protected String[] charset;
	protected String[] separator;
	protected int[] segCountList;
	protected String[] sourcePathPattern;
	protected int[][] keyColumnList;
	protected int[][] valueColumnList;
	
	T mapperEntity;
	SourceRecordFilter sourceRecordFilter;
	
	@SuppressWarnings("unchecked")
	@Override 
	public void configure(JobConf conf){
		//source-count
		sourceCount = conf.getInt(BaseMapredParams.SOURCE_COUNT_PARAM_NAME, 0);
		if (sourceCount <= 0){
			throw new IllegalArgumentException("a positive source-count parameter should be provided");
		}
		//charsets
		charset = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.CHARSETS_PARAM_NAME);
		if (charset == null){
			charset = new String[sourceCount];
			//throw new IllegalArgumentException("charsets parameter must be provided");
		}else if (charset.length != sourceCount){
			throw new IllegalArgumentException("the number of charsets is not source-count");
		}
		//separators
		separator = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.SEPARATORS_PARAM_NAME);
		if (separator == null){
			separator = new String[sourceCount];
			//throw new IllegalArgumentException("separator parameter must be provided");
		}else if (separator.length != sourceCount){
			throw new IllegalArgumentException("the number of separators is not source-count");
		}
		//segcount-list
		segCountList = new int[sourceCount];
		String[] segCountListStrArray = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.SEGCOUNT_LIST_PARAM_NAME);
		if (segCountListStrArray != null){
			if (segCountListStrArray.length != sourceCount){
				throw new IllegalArgumentException("the length of segcount-list is not source-count");
			}else{
				for (int i = 0; i < sourceCount; i++){
					segCountList[i] = Integer.parseInt(segCountListStrArray[i]);
				}
			}
		}
		//source-path-patterns
		sourcePathPattern = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.SOURCE_PATH_PATTERNS_PARAM_NAME);
		if (sourcePathPattern == null){
			throw new IllegalArgumentException("source-path-patterns are not provided.");
		}else if (sourcePathPattern.length != sourceCount){
			throw new IllegalArgumentException("the number of source-path-patterns is not source-count");
		}
		//entity class
		Class<T> mapperEntityClass;
		Type c = ((ParameterizedType)this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
		mapperEntityClass = (Class<T>)c;
		mapperEntity = ReflectionUtils.newInstance(mapperEntityClass, conf);
		//record filter class
		sourceRecordFilter = null;
		String recordFilterClassName = conf.get(BaseMapredParams.SOURCERECORDFILTER_CLASSNAME_PARAM_NAME);
		if (recordFilterClassName != null){
			try {
				SourceRecordFilter filter = ReflectionUtils.newInstance( 
					(Class<SourceRecordFilter>) Class.forName(recordFilterClassName),
					conf);
				sourceRecordFilter = filter;
			} catch (ClassNotFoundException e) {
				throw new IllegalArgumentException("class not found: " + recordFilterClassName);
			}
		}
		//key/value column list
		this.keyColumnList = BaseMapredUtils.getKeyColumnList(sourceCount, conf);
		this.valueColumnList = BaseMapredUtils.getValueColumnListEx(sourceCount, conf, keyColumnList);
	}
	
	@Override
	public void map(LongWritable key, Text value, OutputCollector<K, V> output,
			Reporter reporter) throws IOException {
		
		String inputPath = TaggedInputSplitWrapper.getFileSplit(reporter.getInputSplit()).getPath().toString();	
		
		int sourceIndex = this.getSourceIndex(inputPath);
		if (sourceIndex < 0){
			System.err.println("bad record - can't determine source index.");
			System.err.println("\trecord line: " + value.toString());
			return;
		}
		
		String line;
		if (charset[sourceIndex] == null){
			line = value.toString();
		}else{
			line = new String(value.getBytes(), 0, value.getLength(), charset[sourceIndex]);
		}
		String[] segment;
		if (separator[sourceIndex] != null) {
			String temp_line = line + separator[sourceIndex] + "trouger.random.terminator-PcNUBBGcizvvAhXdMUeQquVzdDHxMdhd";
			//String[] temp_segment = temp_line.split(separator[sourceIndex]);
			String[] temp_segment = BaseMapredUtils.plainTextSplit(temp_line, separator[sourceIndex]);
			int n = temp_segment.length - 1;
			segment = new String[n + 1];
			for (int i = 0; i < n; i++){
				segment[i + 1] = temp_segment[i];
			}
		}else{
			segment = new String[]{null, line};
		}
		if (segCountList[sourceIndex] > 0 && segment.length - 1 < segCountList[sourceIndex]){
			System.err.println("bad record - too less field count.");
			System.err.println("\trecord is from source table " + (sourceIndex + 1));
			System.err.println("\trecord line: " + line);
			System.err.println("\texpected segment count: " + segCountList[sourceIndex]);
			System.err.println("\tactual segment count: " + (segment.length - 1));
			return;
		}

		if (sourceRecordFilter != null && 
		!sourceRecordFilter.isValidRecord(sourceIndex + 1, segment)){
			return;
		}
		if (mapperEntity.acceptLine(sourceIndex, segment)){
			output.collect(mapperEntity.getKey(), mapperEntity.getValue());
		}
	}
	
	/**
	 * Use the path of the input file of the current record
	 * to determine the source index of the record. 
	 * @param inputPath the path of the input file of the current record
	 * @return the zero-based index of the source table of the current record
	 */
	protected int getSourceIndex(String inputPath){
		int ret = -1;
		for (int i = 0; i < sourceCount; i++){
			if (inputPath.contains(sourcePathPattern[i])){
				if (ret < 0) ret = i;
				else return -1;
			}
		}
		return ret;
	}
}
